﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net.Sockets;
using System.Security.Cryptography;
using System.Text;

using Common.Logging;

using NMemcached.Client.Configuration;

namespace NMemcached.Client
{
    /// <summary>
    /// The default IConnectionProvider implementation.
    /// </summary>
    /// <remarks>
    /// <para>The default implementation of the consistent hashing algorithm uses the "ketama" method.</para>
    /// <para>Information on the "ketama" method can be found at "http://chris.de/plugin/tag/ketama"</para>
    /// </remarks>
    public class ConnectionProvider : IConnectionProvider
    {
        private readonly ILog _log = LogManager.GetCurrentClassLogger();

        private const int SERVER_KEY_LIMIT = 100;

        private static readonly IEnumerable<ServerConnection> EmptyList;

        private List<ServerConnection> _allServers;
        private Dictionary<uint, ServerConnection> _serverMap;
        private IEnumerable<ServerConnection> _available;
        private Dictionary<ServerConnection, DateTime> _deadPool;
        private uint[] _continuum;
        private object _syncRoot;
        private bool _disposed;

        /// <summary>
        /// Gets or sets the hash algorithm used to get a server connection
        /// </summary>
        public HashAlgorithm HashAlgorithm { get; private set; }
        /// <summary>
        /// Gets or sets the interval at which dead servers are checked for activity
        /// </summary>
        public TimeSpan PollingInterval { get; set; }

        /// <summary>
        /// Gets a collection of available servers
        /// </summary>
        public IEnumerable<ServerConnection> Servers
        {
            get
            {
                if (this._available == null)
                {
                    return ConnectionProvider.EmptyList;
                }
                else
                {
                    return this._available;
                }
            }
        }

        /// <summary>
        /// Gets a collection of unavailable servers
        /// </summary>
        public IEnumerable<ServerConnection> Unavailable
        {
            get
            {
                if (this._deadPool == null)
                {
                    return ConnectionProvider.EmptyList;
                }
                else
                {
                    return this._deadPool.Keys;
                }
            }
        }

        static ConnectionProvider()
        {
            ConnectionProvider.EmptyList = new ServerConnection[0];
        }

        /// <summary>
        /// Initalizes a new instance of the ConnectionProvider class
        /// </summary>
        /// <param name="servers">A collection of server connections for the provider</param>
        public ConnectionProvider(IEnumerable<ServerConnection> servers)
            : this(servers, null)
        {
        }

        /// <summary>
        /// Initalizes a new instance of the ConnectionProvider class
        /// </summary>
        /// <param name="servers">A collection of server connections for the provider</param>
        /// <param name="hashAlgorithm">The hash algorithm used to hash the keys</param>
        /// <exception cref="System.ArgumentNullException">A null server name was found in the collection</exception>
        public ConnectionProvider(IEnumerable<ServerConnection> servers, HashAlgorithm hashAlgorithm)
        {
            if (servers == null)
            {
                throw new ArgumentNullException("servers");
            }

            if (servers.Count() == 0)
            {
                throw new ArgumentException(StringTable.Get("error_NoServers"), "servers");
            }

            //Clone the server collection to prevent changes done to the source
            //from impacting this list
            this._allServers = new List<ServerConnection>(servers);
            this._serverMap = new Dictionary<uint, ServerConnection>();
            this._available = new List<ServerConnection>();
            this._deadPool = new Dictionary<ServerConnection, DateTime>();
            this._syncRoot = new object();
            this._disposed = false;

            this.PollingInterval = new TimeSpan(0, 5, 0);

            if (hashAlgorithm == null)
            {
                this.HashAlgorithm = new FNVManaged();
            }
            else
            {
                this.HashAlgorithm = hashAlgorithm;
            }

            this._log.Info(m => m(StringTable.Get("log_CreateConnectionProvider"), servers.Count(), this.HashAlgorithm.ToString()));

            try
            {
                this.IndexServers();
            }
            catch (ArgumentNullException e)
            {
                throw e;
            }
        }

        /// <summary>
        /// Closes all server connections
        /// </summary>
        public void Close()
        {
            if (this._allServers != null)
            {
                lock (this._syncRoot)
                {
                    foreach (ServerConnection connection in this._allServers)
                    {
                        try
                        {
                            connection.Close();
                        }
                        catch (Exception e)
                        {
                            this._log.Error(m => m(StringTable.Get("log_ErrorClosingConnection"), e.Message));
                        }
                        finally
                        {
                            connection.Dispose();
                        }
                    }
                }
            }
        }

        /// <summary>
        /// Gets a server connection for the specified key
        /// </summary>
        /// <param name="key">The key of the cache entry</param>
        /// <returns>A server connection for the specified key</returns>
        public ServerConnection GetConnection(string key)
        {
            this.ReinstateDeadServers();

            ServerConnection connection = null;

            uint hash = this.Hash(key);

            //Only bother trying to get a connection if we have available servers
            if (this._available != null && this._available.Count() > 0)
            {
                if (Array.BinarySearch(this._continuum, hash) < 0)
                {
                    //The hash value is not found in the continuum so get
                    //the next highest value
                    uint nextHigher = this._continuum.FirstOrDefault(k => k > hash);
                    //If we get the default back it means there's nothing
                    //higher. Use the first in the list.
                    if (nextHigher == 0)
                    {
                        connection = this._serverMap[this._continuum[0]];
                    }
                    else
                    {
                        connection = this._serverMap[nextHigher];
                    }
                }
                else
                {
                    //The hash value was found so set the connection
                    connection = this._serverMap[hash];
                }
            }

            return connection;
        }

        /// <summary>
        /// Gets the server connections for the specified keys
        /// </summary>
        /// <param name="keys">The keys of the cache entries</param>
        /// <returns>A dictionary of server connection to cache key mappings</returns>
        public IDictionary<ServerConnection, List<string>> GetConnections(params string[] keys)
        {
            Dictionary<ServerConnection, List<string>> serverKeyMap = null;
            if (keys != null && keys.Length > 0)
            {
                serverKeyMap = new Dictionary<ServerConnection, List<string>>();
                ServerConnection current = null;
                foreach (string key in keys)
                {
                    if (!String.IsNullOrEmpty(key))
                    {
                        current = this.GetConnection(key);
                        if (!serverKeyMap.ContainsKey(current))
                        {
                            serverKeyMap.Add(current, new List<string>());
                        }
                        serverKeyMap[current].Add(key);
                    }
                }
            }
            return serverKeyMap;
        }

        /// <summary>
        /// Removes a connection from the list of available servers
        /// </summary>
        /// <param name="connection">The connection to remove</param>
        /// <exception cref="System.ArgumentNullException">A null server name was found in the collection</exception>
        [Obsolete("This method was deprecated in version 1.1 and will not be available in future versions.", false)]
        public void MarkDown(ServerConnection connection)
        {
            //Ensure that only one thread can call this at a time
            //by locking the instance rather than this._syncRoot
            lock (this)
            {
                if (!this._deadPool.ContainsKey(connection))
                {
                    this._deadPool.Add(connection, DateTime.Now);
                    try
                    {
                        this.IndexServers();
                    }
                    catch (ArgumentNullException e)
                    {
                        throw e;
                    }
                }
            }
        }

        /// <summary>
        /// Removes a connection from the list of available servers
        /// </summary>
        /// <param name="connection">The connection to remove</param>
        /// <exception cref="System.ArgumentNullException">A null server name was found in the collection</exception>
        public void DisableConnection(ServerConnection connection)
        {
            //Ensure that only one thread can call this at a time
            //by locking the instance rather than this._syncRoot
            lock (this)
            {
                if (!this._deadPool.ContainsKey(connection))
                {
                    this._deadPool.Add(connection, DateTime.Now);
                    try
                    {
                        this.IndexServers();
                    }
                    catch (ArgumentNullException e)
                    {
                        throw e;
                    }
                }
            }
        }

        /// <summary>
        /// Releases the unmanaged resources used by the ConnectionProvider
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        //Protected methods

        /// <summary>
        /// Releases the unmanaged resources used by the ConnectionProvider, and optionally disposes of the managed resources
        /// </summary>
        /// <param name="disposing">true to release both managed and unmanaged resources; false to releases only unmanaged resources</param>
        protected virtual void Dispose(bool disposing)
        {
            if (!this._disposed)
            {
                if (disposing)
                {
                }

                if (this._allServers != null)
                {
                    lock (this._syncRoot)
                    {
                        foreach (IDisposable server in this._allServers)
                        {
                            server.Dispose();
                        }
                    }
                }

                this._disposed = true;
            }
        }

        //Private methods

        private uint Hash(string data)
        {
            if (String.IsNullOrEmpty(data))
            {
                throw new ArgumentNullException("data");
            }
            byte[] dataBytes = Encoding.UTF8.GetBytes(data);
            //Creating an integer from the first 32 bits of the hash
            //result can result in collisions. However, collisions are
            //acceptable since the client is trying for a consistent
            //hashing system to determine the locaiton of cached items.
            return BitConverter.ToUInt32(this.HashAlgorithm.ComputeHash(dataBytes), 0);
        }

        private void IndexServers()
        {
            //This class lock is done intentionally to ensure that only one
            //thread can call this at a time.
            lock (this)
            {
                //Set the the available server list
                if (this._allServers != null && this._deadPool != null)
                {
                    this._available = this._allServers.Except(this._deadPool.Keys);
                }
                else
                {
                    this._available = ConnectionProvider.EmptyList;
                }

                //Clear the server map
                this._serverMap.Clear();
                //Create the continuum
                this._continuum = new uint[this._available.Count() * ConnectionProvider.SERVER_KEY_LIMIT];

                //The server map and continuum are empty at this point because
                //the available count is zero. Don't continue to try to index the
                //servers.
                if (this._available.Count() > 0)
                {
                    //Create a hash set to hold the keys
                    HashSet<uint> allKeys = new HashSet<uint>();
                    //Create a buffer for the generated keys
                    uint[] keys = new uint[ConnectionProvider.SERVER_KEY_LIMIT];
                    //Initialize the hash and the increment
                    uint hash = 0;
                    int increment = 0;
                    //Create SERVER_KEY_LIMIT keys for each server
                    foreach (ServerConnection server in this._available)
                    {
                        for (int i = 0; i < keys.Length; i++)
                        {
                            //Create the hash from the server plus the increment value
                            hash = this.Hash(String.Concat(server.ToString(), i.ToString()));
                            //Create the hash until it does not already exist
                            //This will happen more frequently if using a hashing algorithm that
                            //returns a value greater than 32 bits.
                            while (allKeys.Contains(hash))
                            {
                                hash = this.Hash(String.Concat(server.ToString(), i.ToString(), (++increment).ToString()));
                            }
                            allKeys.Add(hash);
                            this._serverMap.Add(hash, server);
                            //Reset the increment value for the next iteration
                            increment = 0;
                        }
                    }

                    allKeys.CopyTo(this._continuum);
                    Array.Sort(this._continuum);

                    this._log.Info(m => m(StringTable.Get("log_IndexedServerCount"), this._available.Count()));
                }
            }
        }

        //Moves connections out of the dead server list after the specified polling interval.
        //If the server is still unavailable, the time of death will be updated so it is
        //tried again the next time the interval passes.
        private void ReinstateDeadServers()
        {
            if (this._deadPool != null && this._deadPool.Keys.Count > 0)
            {
                //This lock is on purpose. This entire instance needs to be inaccessable
                //to all threads during this process.
                lock (this)
                {
                    /*
                     * Get the connections to reinstate.
                     * 
                     * Example:
                     * 
                     *  this.PollingInterval = 5 mintues
                     *  A server errors and is marked dead at 12:00
                     *  At 12:01 the server will not be checked because 12:00 + 5 minutes (12:05) is greater than the current time
                     *  At 12:07 the server will be checked because 12:00 + 5 minutes (12:05) is less than the current time
                     */

                    //Make a copy of the connections to be reinstated
                    //so we're not trying to modify a collection being enumerated

                    List<ServerConnection> connections = new List<ServerConnection>();

                    if (this._deadPool != null)
                    {
                        connections.AddRange(from connection in this._deadPool
                                             where connection.Value.Add(this.PollingInterval) < DateTime.Now
                                             select connection.Key);
                    }
                    else
                    {
                        connections = new List<ServerConnection>();
                    }

                    int reinstateCount = 0;
                    //Take the servers out of the dead pool that are active again
                    foreach (ServerConnection con in connections)
                    {
                        try
                        {
                            //Get the version from the server to check the connection
                            //This will throw a socket exception if it's still dead
                            con.Version();
                            //Remove the connection from the dead pool
                            if (this._deadPool.Remove(con))
                            {
                                //Keep track of how many servers are reinstated
                                reinstateCount++;
                            }
                        }
                        catch (SocketException)
                        {
                            //Ignore any socket exceptions. The server is still dead
                            //so update the time of death and try after the next
                            //interval has passed
                            this._deadPool[con] = DateTime.Now;
                        }
                    }

                    //Reindex the continuum if any of the servers were reinstated
                    if (reinstateCount > 0)
                    {
                        this._log.Info(m => m(StringTable.Get("log_ReinstatedServerCount"), reinstateCount));
                        this.IndexServers();
                    }
                }
            }
        }
    }
}